package com.atbeijing.gmall.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atbeijing.gmall.realtime.utils.DimUtil;
import com.atbeijing.gmall.realtime.utils.ThreadPoolUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.util.Arrays;
import java.util.concurrent.ExecutorService;

/**
 * 异步维度关联函数类
 * 模板方法设计模式
 * 在父类中定义实现某一个功能的核心算法的骨架，将具体实现延迟到子类中
 * 在不改变父类中核心算法骨架前提下，每一个子类中都可以有自己不同的实现。
 */
public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T,T> implements DimJoinFunction<T> {
    //线程池
    private ExecutorService executorService;

    //表名
    private String tableName;

    public DimAsyncFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        executorService = ThreadPoolUtil.getInstance();
    }

    /**
     * 数据流中每条数据执行一次该方法
     * @param input 数据流
     * @param resultFuture 结果集
     * @throws Exception
     */
    @Override
    public void asyncInvoke(T input, ResultFuture<T> resultFuture) throws Exception {
        //利用线程池中的线程执行异步操作
        executorService.submit(new Runnable() {
            @Override
            public void run() {
                long start = System.currentTimeMillis();
                //模板方法设计模式  获取查询维度表所需主键
                String key = getKey(input);
                //从缓存或者hbase获取维度对象
                JSONObject dimInfo = DimUtil.getDimInfo(tableName, key);
                if (dimInfo!=null && dimInfo.size()>0){
                    //模板方法设计模式  宽表与维度表关联
                    join(input,dimInfo);
                }
                long end = System.currentTimeMillis();
                System.out.println("异步维度表"+tableName+"关联耗时:" +(end - start) +"毫秒");
                //接收异步响应结果，向下游传递数据
                resultFuture.complete(Arrays.asList(input));
            }
        });
    }



}
